﻿using System;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading;
using System.Xml;
using Common.Utilities;
using Common.Utilities.Dispatcher;

namespace Workflows.Components.Entities.MasterData
{
    public class FieldInteractionWorker:IWorker 
    {
        #region Implementation of IWorker

        public event WorkerStarted WorkerStartedEvent;
        public event WorkerFinished WorkerFinishedEvent;
        public event WorkerHasError WorkerErredEvent;
        public event WorkerCancelled WorkerCancelledEvent;
        public event WorkerStatusChanged WorkerStatusChangedEvent;

        private Thread workerThread;
        private FieldInteractionInstruction instruction;
        private FieldInteractionOutput output;
        private RunContext _Context;
        private Guid _TicketID;
        private string _WorkflowName;
        private string _ComponentName;

        /// <summary>
        /// 
        /// </summary>
        public string ComponentName
        {
            get { return _ComponentName; }
        }

        /// <summary>
        /// 
        /// </summary>
        public string WorkflowName
        {
            get { return _WorkflowName; }
        }

        /// <summary></summary>
        public Thread WorkerThread
        {
            get { return workerThread; }
        }
        /// <summary></summary>
        public InstructionBase Instruction
        {
            get { return instruction; }
        }
        /// <summary></summary>
        public ResultBase Output
        {
            get { return output; }
        }
        /// <summary></summary>
        public Type ComponentType
        {
            get { return typeof(FieldInteractionComponent); }
        }
        /// <summary></summary>
        public RunContext Context
        {
            get { return _Context; }
            set { _Context = value; }
        }
        /// <summary></summary>
        public void SetInstruction(Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._TicketID = ticketGuid;
            this.instruction = (FieldInteractionInstruction)request;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void SetInstruction(RunContext context, Guid ticketGuid, InstructionBase request, string workflowName, string componentName)
        {
            this._TicketID = ticketGuid;
            this.instruction = (FieldInteractionInstruction)request;
            this._Context = context;
            this._WorkflowName = workflowName;
            this._ComponentName = componentName;
        }
        /// <summary></summary>
        public void DoWork()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
            }
            this.workerThread = Thread.CurrentThread;
            this.CalculateFieldDependency();
        }
        /// <summary></summary>
        public void DoWorkAsync()
        {
            if (this.WorkerStartedEvent != null)
            {
                this.WorkerStartedEvent(this._TicketID, this, this.GetType(), this.instruction);
            }
            this.workerThread = new Thread(new ThreadStart(this.CalculateFieldDependency));
            this.workerThread.Start();
        }
        /// <summary></summary>
        public void CancelWork()
        {
            if (this.workerThread != null && this.workerThread.IsAlive)
            {
                this.workerThread.Abort();
                this.workerThread.Join();

                if (this.WorkerCancelledEvent != null)
                {
                    this.WorkerCancelledEvent(this._TicketID, this.GetType(), this.instruction);
                }
            }
        }

        #endregion

        #region work
        private void CalculateFieldDependency()
        {
            try
            {
                int recCount = 0;
                string sql = string.Empty;
                if (!string.IsNullOrEmpty(this.instruction.Sql))
                {
                    sql = string.Format("select count (distinct tmp.{0}) from ({1}) as tmp where tmp.{0} is not null",
                                        this.instruction.DeterminantField, this.instruction.Sql);
                }
                else
                {
                    sql = string.Format("select count (distinct {0}) from {1}.{2} where {0} is not null",
                                        this.instruction.DeterminantField, this.instruction.SchemaName,
                                        this.instruction.TblName);
                }
                recCount = DbUtil.GetTableRecordCount(DataProviderType.MSSQL, this.instruction.ConnStr,
                                                      sql);
                Dictionary<string, double> correspondenceScores =
                    this.GetDependentFieldsCorrespondenceScores(recCount);
                Dictionary<string, double> correlationScores = 
                    this.GetFieldCorrelationScores();
                Dictionary<string, double> rScores = this.GetFieldRScores();
                XmlDocument xDoc=new XmlDocument();
                xDoc.LoadXml("<FieldDependencies></FieldDependencies>");
                XmlNode root = xDoc.DocumentElement;
                XmlDataUtil.UpdateAttribute(ref xDoc, root, "ConnStr", this.instruction.ConnStr);
                XmlDataUtil.UpdateAttribute(ref xDoc, root, "SchemaName", this.instruction.SchemaName);
                XmlDataUtil.UpdateAttribute(ref xDoc, root, "TblName", this.instruction.TblName);
                XmlDataUtil.UpdateAttribute(ref xDoc, root, "Sql", this.instruction.Sql);
                XmlDataUtil.UpdateAttribute(ref xDoc, root,"Determinant",this.instruction.DeterminantField);
                XmlDataUtil.UpdateAttribute(ref xDoc, root, "Count", recCount.ToString());
                foreach(string fieldName in this.instruction.DependentFields)
                {
                    XmlNode dependentNode = XmlDataUtil.AddElement(ref xDoc, root, "DependentField");
                    XmlDataUtil.UpdateAttribute(ref xDoc, dependentNode,"FieldName", fieldName);
                    if(correspondenceScores!=null && correspondenceScores.ContainsKey(fieldName))
                        XmlDataUtil.UpdateAttribute(ref xDoc, dependentNode,"Correspondence",correspondenceScores[fieldName].ToString());
                    else 
                        XmlDataUtil.UpdateAttribute(ref xDoc, dependentNode, "Correspondence", "0");
                    if(correlationScores !=null && correlationScores.ContainsKey(fieldName))
                        XmlDataUtil.UpdateAttribute(ref xDoc, dependentNode, "Correlation", correlationScores[fieldName].ToString());
                    else 
                        XmlDataUtil.UpdateAttribute(ref xDoc, dependentNode, "Correlation", "0");
                    if(rScores !=null && rScores.ContainsKey(fieldName))
                    {
                        XmlDataUtil.UpdateAttribute(ref xDoc, dependentNode, "RScore", rScores[fieldName].ToString());
                        XmlDataUtil.UpdateAttribute(
                            ref xDoc, dependentNode, "R2Score", Math.Pow(rScores[fieldName], 2).ToString());
                    }
                    else
                    {
                        XmlDataUtil.UpdateAttribute(ref xDoc, dependentNode, "RScore", "0");
                        XmlDataUtil.UpdateAttribute(ref xDoc, dependentNode, "R2Score", "0");
                    }
                }
                xDoc.Save(this.instruction.OutputFilePath);

                this.output = new FieldInteractionOutput(this.instruction.OwnerComponentID, this._TicketID,
                                                        this.instruction.OutputFilePath);
                if (this.WorkerFinishedEvent != null)
                    this.WorkerFinishedEvent(this._TicketID, this.GetType(), this.instruction, this.output);
            }
            catch(Exception ex)
            {
                if (this.WorkerErredEvent != null)
                    this.WorkerErredEvent(this._TicketID, this.GetType(), this.instruction, ex.ToString(), true);
            }
        }
        #endregion

        #region correspondence
        private Dictionary<string,double> GetDependentFieldsCorrespondenceScores(
            int recordCount)
        {
            Dictionary<string,double> fieldCorrespondenceScores=new Dictionary<string, double>();
            foreach(string dependentFieldName in this.instruction.DependentFields)
            {
                string sql = string.Empty;
                string sql2 = string.Empty;
                if(string.IsNullOrEmpty(this.instruction.Sql))
                {
                    sql = string.Format(
                        "select count (distinct {0}) from {1}.{2} where {0} is not null",
                        dependentFieldName, this.instruction.SchemaName, this.instruction.TblName);
                    sql2 = string.Format(
                        "select count (distinct t1.{0}) from {1}.{2} t1 group by {3} having count (distinct {0})>1", 
                        dependentFieldName, 
                        this.instruction.SchemaName, 
                        this.instruction.TblName,
                        this.instruction.DeterminantField);
                }
                else
                {
                    sql = string.Format(
                        "select count (distinct tmp.{0}) from ({1}) as tmp where tmp.{0} is not null",
                        dependentFieldName, this.instruction.Sql);
                    sql2 = string.Format(
                        "select count (distinct tmp.{0}) from ({1}) as tmp group by tmp.{2} having count (distinct tmp.{0})>1",
                        dependentFieldName, this.instruction.Sql, this.instruction.DeterminantField);
                }
                int distCount =
                    SimpleDataTypeUtil.ReadInt(DbUtil.ExecuteScalar(DataProviderType.MSSQL, this.instruction.ConnStr,
                                                                    sql));
                int descrepencyCount =
                    SimpleDataTypeUtil.ReadInt(DbUtil.ExecuteScalar(DataProviderType.MSSQL, this.instruction.ConnStr,
                                                                    sql2));
                double score = 1 - (double)descrepencyCount / recordCount;
                fieldCorrespondenceScores.Add(dependentFieldName, score);
            }
            return fieldCorrespondenceScores;
        }
        #endregion

        #region correlation
        /// <summary>
        /// only number fields
        /// </summary>
        /// <param name="threshold"></param>
        /// <returns></returns>
        private Dictionary<string,double> GetFieldCorrelationScores()
        {
            TableColumn[] tblCols = null;
            if (!string.IsNullOrEmpty(this.instruction.Sql))
                tblCols = DbUtil.GetTableColumns(this.instruction.ConnStr, this.instruction.Sql);
            else
                tblCols = DbUtil.GetTableColumns(DataProviderType.MSSQL, this.instruction.ConnStr,
                                                 this.instruction.SchemaName, this.instruction.TblName);
            if(tblCols==null)
                throw new Exception("Unable to retrieve schema");
            Dictionary<string,SqlDbType> colDbTypes=new Dictionary<string, SqlDbType>();
            foreach(TableColumn col in tblCols)
            {
                colDbTypes.Add(col.ColumnName, col.DbType);
                if(col.ColumnName.ToLower()==this.instruction.DeterminantField.ToLower())
                {
                    if(!FieldDataType.IsNumber(col.DbType))
                        return new Dictionary<string, double>();
                }
            }

            Dictionary<string,double> fieldCorrelationScores=new Dictionary<string, double>();
            foreach(string dependentField in this.instruction.DependentFields)
            {
                if(!FieldDataType.IsNumber(colDbTypes[dependentField]))
                {
                    fieldCorrelationScores.Add(dependentField, 0);
                }
                else
                {
                    double score = 0;
                    string sql = string.Empty;
                    if (string.IsNullOrEmpty(this.instruction.Sql))
                    {
                        sql = string.Format(
                            "select {0} as p, {1} as d from {2}.{3} where {0} is not null and {1} is not null",
                            dependentField, this.instruction.DeterminantField,
                            this.instruction.SchemaName, this.instruction.TblName);
                    }
                    else
                    {
                        sql = string.Format(
                            "select tmp.{0} as p, tmp.{1} as d from ({2}) as tmp where tmp.{0} is not null and tmp.{1} is not null",
                            dependentField, this.instruction.DeterminantField, this.instruction.Sql);
                    }
                    DataTable dt = DbUtil.GetTableDataBySQL(this.instruction.ConnStr, "Data", sql);
                    if (dt == null || dt.Rows.Count == 0)
                        score = 0;
                    else
                    {
                        int num = dt.Rows.Count;
                        float sumProd = 0;
                        float sum1 = 0;
                        float sum2 = 0;
                        float sumSq1 = 0;
                        float sumSq2 = 0;
                        foreach (DataRow dr in dt.Rows)
                        {
                            double d1 = double.Parse(dr["p"].ToString());
                            double d2 = double.Parse(dr["d"].ToString());
                            float f1 = (float) d1;
                            float f2 = (float) d2;

                            sumProd += f1*f2;
                            sum1 += f1;
                            sum2 += f2;
                            sumSq1 += f1*f1;
                            sumSq2 += f2*f2;
                        }
                        double top = num*sumProd - sum1*sum2;
                        double bottom1 = num*sumSq1 - sum1*sum1;
                        double bottom2 = num*sumSq2 - sum2*sum2;
                        double bottom = Math.Sqrt(bottom1*bottom2);
                        score = top/bottom;
                    }
                    fieldCorrelationScores.Add(dependentField, score);
                }
            }
            return fieldCorrelationScores;
        }
        #endregion

        #region Correlation Coefficient
        /// <summary>
        /// also called Correlation coefficient, r score
        /// The quantity r, called the linear correlation coefficient, measures the strength and 
        /// the direction of a linear relationship between two variables. The linear correlation
        /// coefficient is sometimes referred to as the Pearson product moment correlation coefficient in
        /// honor of its developer Karl Pearson.
        /// </summary>
        /// <returns></returns>
        private Dictionary<string,double> GetFieldRScores()
        {
            TableColumn[] tblCols = null;
            if (!string.IsNullOrEmpty(this.instruction.Sql))
                tblCols = DbUtil.GetTableColumns(this.instruction.ConnStr, this.instruction.Sql);
            else
                tblCols = DbUtil.GetTableColumns(DataProviderType.MSSQL, this.instruction.ConnStr,
                                                 this.instruction.SchemaName, this.instruction.TblName);
            if (tblCols == null)
                throw new Exception("Unable to retrieve schema");
            Dictionary<string, SqlDbType> colDbTypes = new Dictionary<string, SqlDbType>();
            foreach (TableColumn col in tblCols)
            {
                colDbTypes.Add(col.ColumnName, col.DbType);
                if (col.ColumnName.ToLower() == this.instruction.DeterminantField.ToLower())
                {
                    if (!FieldDataType.IsNumber(col.DbType))
                        return new Dictionary<string, double>();
                }
            }

            Dictionary<string, double> rScores = new Dictionary<string, double>();
            foreach (string dependentField in this.instruction.DependentFields)
            {
                if (!FieldDataType.IsNumber(colDbTypes[dependentField]))
                {
                    rScores.Add(dependentField, 0);
                }
                else
                {
                    double score = 0;
                    string sql = string.Empty;
                    if (string.IsNullOrEmpty(this.instruction.Sql))
                    {
                        sql = string.Format(
                            "select {0} as p, {1} as d from {2}.{3} where {0} is not null and {1} is not null",
                            dependentField, this.instruction.DeterminantField,
                            this.instruction.SchemaName, this.instruction.TblName);
                    }
                    else
                    {
                        sql = string.Format(
                            "select tmp.{0} as p, tmp.{1} as d from ({2}) as tmp where tmp.{0} is not null and tmp.{1} is not null",
                            dependentField, this.instruction.DeterminantField, this.instruction.Sql);
                    }
                    DataTable dt = DbUtil.GetTableDataBySQL(this.instruction.ConnStr, "Data", sql);
                    if (dt == null || dt.Rows.Count == 0)
                        score = 0;
                    else
                    {
                        int num = dt.Rows.Count;
                        float sumProd = 0;
                        float sum1 = 0;
                        float sum2 = 0;
                        float sumSq1 = 0;
                        float sumSq2 = 0;
                        foreach (DataRow dr in dt.Rows)
                        {
                            double d1 = double.Parse(dr["p"].ToString());
                            double d2 = double.Parse(dr["d"].ToString());
                            float f1 = (float)d1;
                            float f2 = (float)d2;

                            sumProd += f1 * f2;
                            sum1 += f1;
                            sum2 += f2;
                            sumSq1 += f1 * f1;
                            sumSq2 += f2 * f2;
                        }
                        double top = num * sumProd - sum1 * sum2;
                        double bottom1 = num * sumSq1 - sum1 * sum1;
                        bottom1 = Math.Sqrt(bottom1);
                        double bottom2 = num * sumSq2 - sum2 * sum2;
                        bottom2 = Math.Sqrt(bottom2);
                        double bottom = bottom1 * bottom2;
                        score = top / bottom;
                    }
                    rScores.Add(dependentField, score);
                }
            }
            return rScores;
        }

        /// <summary>
        /// also called coefficient of determination (r2)
        /// is useful because it gives the proportion of 
        /// the variance (fluctuation) of one variable that is predictable from the other variable.
        /// It is a measure that allows us to determine how certain one can be in making
        /// predictions from a certain model/graph
        /// </summary>
        /// <param name="threshold"></param>
        /// <returns></returns>
        private Dictionary<string,double> GetFieldRSquareScores(double threshold)
        {
            TableColumn[] tblCols = null;
            if (!string.IsNullOrEmpty(this.instruction.Sql))
                tblCols = DbUtil.GetTableColumns(this.instruction.ConnStr, this.instruction.Sql);
            else
                tblCols = DbUtil.GetTableColumns(DataProviderType.MSSQL, this.instruction.ConnStr,
                                                 this.instruction.SchemaName, this.instruction.TblName);
            if (tblCols == null)
                throw new Exception("Unable to retrieve schema");
            Dictionary<string, SqlDbType> colDbTypes = new Dictionary<string, SqlDbType>();
            foreach (TableColumn col in tblCols)
            {
                colDbTypes.Add(col.ColumnName, col.DbType);
                if (col.ColumnName.ToLower() == this.instruction.DeterminantField.ToLower())
                {
                    if (!FieldDataType.IsNumber(col.DbType))
                        return new Dictionary<string, double>();
                }
            }

            Dictionary<string, double> r2Scores = new Dictionary<string, double>();
            foreach (string dependentField in this.instruction.DependentFields)
            {
                if (!FieldDataType.IsNumber(colDbTypes[dependentField]) &&
                    !FieldDataType.IsDateTime(colDbTypes[dependentField]))
                {
                    r2Scores.Add(dependentField, 0);
                }
                else
                {
                    double score = 0;
                    string sql = string.Empty;
                    if (string.IsNullOrEmpty(this.instruction.Sql))
                    {
                        sql = string.Format(
                            "select {0} as p, {1} as d from {2}.{3} where {0} is not null and {1} is not null",
                            dependentField, this.instruction.DeterminantField,
                            this.instruction.SchemaName, this.instruction.TblName);
                    }
                    else
                    {
                        sql = string.Format(
                            "select tmp.{0} as p, tmp.{1} as d from ({2}) as tmp where tmp.{0} is not null and tmp.{1} is not null",
                            dependentField, this.instruction.DeterminantField, this.instruction.Sql);
                    }
                    DataTable dt = DbUtil.GetTableDataBySQL(this.instruction.ConnStr, "Data", sql);
                    if (dt == null || dt.Rows.Count == 0)
                        score = 0;
                    else
                    {
                        int num = dt.Rows.Count;
                        float sumProd = 0;
                        float sum1 = 0;
                        float sum2 = 0;
                        float sumSq1 = 0;
                        float sumSq2 = 0;
                        foreach (DataRow dr in dt.Rows)
                        {
                            double d1 = double.Parse(dr["p"].ToString());
                            double d2 = double.Parse(dr["d"].ToString());
                            float f1 = (float)d1;
                            float f2 = (float)d2;

                            sumProd += f1 * f2;
                            sum1 += f1;
                            sum2 += f2;
                            sumSq1 += f1 * f1;
                            sumSq2 += f2 * f2;
                        }
                        double top = num * sumProd - sum1 * sum2;
                        double bottom1 = num * sumSq1 - sum1 * sum1;
                        bottom1 = Math.Sqrt(bottom1);
                        double bottom2 = num * sumSq2 - sum2 * sum2;
                        bottom2 = Math.Sqrt(bottom2);
                        double bottom = bottom1 * bottom2;
                        score = top / bottom;
                        score = score*score;
                    }
                    if (score >= threshold)
                        r2Scores.Add(dependentField, score);
                    else
                        r2Scores.Add(dependentField, 0);
                }
            }
            return r2Scores;
        }
        #endregion
    }
}
